package offset_gaobo.util;

import kafka.common.TopicAndPartition;
import offset_gaobo.Const;
import offset_gaobo.Offset;
import org.apache.spark.streaming.kafka.OffsetRange;
import org.apache.zookeeper.data.Stat;
import scala.Long;
import scala.collection.mutable.HashMap;
import scala.collection.mutable.Map;

import java.util.List;

public class OffSetUtils {


    //创建工具类 遍历offsetRanges  并保存offset到zookeeper
    public static void saveOffset(OffsetRange[] offsetRanges) {

        int length = offsetRanges.length;

        for (int i = 0; i < length; i++) {
            /*
            val topic: String,
            val partition: Int,
            val fromOffset: Long,
            val untilOffset: Long
             */
            OffsetRange offsetRange = offsetRanges[i];
            /*
                /offset/topic/partition
             */
            //    /topic/partition
            String nodePath = offsetRange.topic() + "/" + offsetRange.partition();
            //todo  判断是否有 该nodePath节点

            try {
                boolean isExist = ZookeeperUtils.checkExists(nodePath);

                if (isExist) {
                    //1.获取state-->version
                    Stat stat = ZookeeperUtils.getStat(nodePath);
                    //2.根据version跟新数据
                    long untilOffset = offsetRange.untilOffset();
                    int version = stat.getVersion();

                    //{"topic":"offsetTopic","partition":"0","from":"100","until":"200"}

                    Offset offset = new Offset();
                    offset.setTopic(offsetRange.topic());
                    offset.setPartition(String.valueOf(offsetRange.partition()));
                    offset.setFrom(String.valueOf(offsetRange.fromOffset()));
                    offset.setUntil(String.valueOf(offsetRange.untilOffset()));
                    String offsetJson = JsonUtils.objectToJson(offset);


                    ZookeeperUtils.updataOffset(nodePath, offsetJson, version);
                } else {
                    //1.创建节点，并写入数据
                    Offset offset = new Offset();
                    offset.setTopic(offsetRange.topic());
                    offset.setPartition(String.valueOf(offsetRange.partition()));
                    offset.setFrom(String.valueOf(offsetRange.fromOffset()));
                    offset.setUntil(String.valueOf(offsetRange.untilOffset()));

                    String offsetJson = JsonUtils.objectToJson(offset);
                    ZookeeperUtils.createNode(nodePath, offsetJson);
                }


            } catch (Exception e) {
                e.printStackTrace();
            }


        }


    }

    // TopicAndPartition(topic, 0) -> ZookeeperUtils.getOffset("/p0")
    public static Map<TopicAndPartition, Long> getOffsetMap() {
        Map offsetMap = new HashMap<TopicAndPartition, Long>();

        //1.到zookeeper拉取 offset信息
        //判断跟节点是否有子节点
        List<String> childrenList = ZookeeperUtils.getChildren(Const.ZOOKEEPER_NAME_SPACE);

        if (childrenList.size() > 0) {
            //2.封装成map
            //如果有子节点，获取zookeeper数据

            for (int i = 0; i < childrenList.size(); i++) {
                //  yy_topic/0
                String nodePath = childrenList.get(i);

                //{"topic":"offsetTopic","partition":"0","from":"100","until":"200"}
                String offsetJson = ZookeeperUtils.getDate(nodePath);
                Offset offset = JsonUtils.jsonToPojo(offsetJson, Offset.class);


                String topic = nodePath.split("/")[0];
                String partition = nodePath.split("/")[1];
                TopicAndPartition topicAndPartition = new TopicAndPartition(topic, Integer.parseInt(partition));
                offsetMap.put(topicAndPartition, java.lang.Long.parseLong(offset.getUntil()));
            }


            //3.返回map
            return offsetMap;
        } else {
            //如果没有子节点，返回空map
            return offsetMap;
        }







    }


}
